// Copyright (C) GridGain Systems, Inc. Licensed under GPLv3, http://www.gnu.org/licenses/gpl.html

/*  _________        _____ __________________        _____
 *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
 *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
 *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
 *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
 */

package grid;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.gridgain.grid.GridException;
import org.gridgain.grid.GridJob;
import org.gridgain.grid.GridJobAdapterEx;
import org.gridgain.grid.GridJobResult;
import org.gridgain.grid.GridNode;
import org.gridgain.grid.GridTaskAdapter;
import org.gridgain.grid.GridTaskContinuousMapper;
import org.gridgain.grid.GridTaskNoResultCache;
import org.gridgain.grid.resources.GridTaskContinuousMapperResource;
import controller.Job;
import controller.Task;



/**
 * This task demonstrates how continuous mapper is used. The passed in phrase is
 * split into multiple words and next word is sent out for processing only when
 * the result for the previous word was received.
 * <p>
 * Note that annotation {@link GridTaskNoResultCache} is optional and tells
 * GridGain not to accumulate results from individual jobs. In this example we
 * increment total character count directly in
 * {@link #result(GridJobResult, List)} method, and therefore don't need to
 * accumulate them be be processed at reduction step.
 * 
 * @author 2005-2011 Copyright (C) GridGain Systems, Inc.
 * @version 3.0.5c.25012011
 */
@GridTaskNoResultCache
public class GridContinuousMapperTask extends
		GridTaskAdapter<Job, Integer> {
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	Task task = null;//new Task(0);

	public GridContinuousMapperTask() throws Exception {
		task = new Task();
		task.loadingList();
		task.generateFrequency();
	}

	// This field will be injected with task continuous mapper.
	@GridTaskContinuousMapperResource
	private GridTaskContinuousMapper mapper;

	/** Word queue. */
	private final Queue<Job> jobs = new ConcurrentLinkedQueue<Job>();

	/** Total character count. */
	// private final AtomicInteger totalChrCnt = new AtomicInteger(0);

	// private JobParams job;
	/** {@inheritDoc} */
	@Override
	public Map<? extends GridJob, GridNode> map(List<GridNode> arg0,
			Job job) throws GridException {
		// this.job = job;
		// if (F.isEmpty(phrase)) {
		// throw new GridException("Phrase is sempty.");
		// }

		//int nbrOfNodes = arg0.size();
		

		// Populate word queue.
		int size = task.getMap().size();
		for (int i = 0; i < size; i++) {
			Collections.addAll(jobs, task.getJob());
		}
		
		for (int i = 0;  i < size; i++) {
			System.out.println("Send Word "+sendWord());
		}

		// Sends first word.

		// Since we have send at least one job, we are allowed to return
		// 'null' from map method.
		return null;
	}

	/** {@inheritDoc} */
	@Override
	public Integer reduce(List<GridJobResult> results) throws GridException {
		return 1; // check it
	}

	/**
	 * Sends next queued word to the next node implicitly selected by load
	 * balancer.
	 * 
	 * @return {@code True} if next word was sent, {@code false} if there are no
	 *         more words to send.
	 * @throws GridException
	 *             If sending of a word failed.
	 */
	private boolean sendWord() throws GridException {
		// Remove first word from the queue.
		
		if (jobs != null) {
			// System.out.println("passei");
			// Map next word.
			

			mapper.send(new GridJobAdapterEx(jobs.poll()) {
				/**
				 * 
				 */
				private static final long serialVersionUID = 1L;

				@Override
				public Object execute() {
					final Job job = argument(0);
					job.execute();
					return 1;
				}
			});

			return true;
		}

		// No more words to map.
		return false;
	}
}
